package WindowFun;

import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.text.SimpleDateFormat;

public class Pro_WindowRes implements WindowFunction<Tuple2<Long,Double>, ClassObject.ProvinceMessage, Tuple, TimeWindow> {
    //定义一个时间格式化工具用来将当前时间(双十一那天订单的时间)转为String格式
    SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    @Override
    public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<Long,Double>> input, Collector<ClassObject.ProvinceMessage> out) throws Exception {
        String address = ((Tuple1<String>) tuple).f0;
        Tuple2<Long,Double> t = input.iterator().next();
        //统计销量金额
        //为了后面项目铺垫,使用一下用Bigdecimal来表示精确的小数
        BigDecimal bigDecimal = new BigDecimal(t.f1);
        //setScale设置精度保留2位小数,
        double roundPrice = bigDecimal.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();//四舍五入
        //统计销量数
        long num = t.f0;

        long currentTimeMillis = System.currentTimeMillis();
        String dateTime = df.format(currentTimeMillis);

        ClassObject.ProvinceMessage res = new ClassObject.ProvinceMessage(address,dateTime,num,roundPrice);
        out.collect(res);
    }
}
